package cep;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * 匹配登录 后操作的
 * 登录后100秒未操作的按超时 侧输出
 */
public class TestCep06 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStreamSource<String> ds = env.readTextFile("data/cep_04.txt");
        KeyedStream<TimeoutVo, String> ds1 = ds.map(new MapFunction<String, TimeoutVo>() {
            @Override
            public TimeoutVo map(String value) throws Exception {
                String[] split = value.split("[_| ]");
                TimeoutVo timeoutVo = new TimeoutVo(Integer.valueOf(split[0]), split[1], split[2], Long.valueOf(split[3]));
                return timeoutVo;
            }
        })
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<TimeoutVo>(Time.seconds(1)) {
                    @Override
                    public long extractTimestamp(TimeoutVo element) {
                        return element.time * 1000;
                    }
                }).keyBy(new KeySelector<TimeoutVo, String>() {
                    @Override
                    public String getKey(TimeoutVo value) throws Exception {
                        return value.name;
                    }
                });//KeyBy必须放在最后

        Pattern<TimeoutVo, TimeoutVo> pattern = Pattern.<TimeoutVo>begin("start").where(new SimpleCondition<TimeoutVo>() {
            @Override
            public boolean filter(TimeoutVo value) throws Exception {
                return value.op.equals("login");
            }
        }).next("next").where(new SimpleCondition<TimeoutVo>() {
            @Override
            public boolean filter(TimeoutVo value) throws Exception {
                return value.op.equals("operator");
            }
        }).within(Time.seconds(10));

        PatternStream<TimeoutVo> ps = CEP.pattern(ds1, pattern);

        OutputTag<TimeoutVo> tag = new OutputTag<>("late", TypeInformation.of(TimeoutVo.class));

        SingleOutputStreamOperator<String> res = ps.select(tag, new PatternTimeoutFunction<TimeoutVo, TimeoutVo>() {
            @Override//延迟处理函数
            public TimeoutVo timeout(Map<String, List<TimeoutVo>> map, long l) throws Exception {
                List<TimeoutVo> next = map.get("start");
                if (next != null && next.size() > 0)
                    return next.get(0);//分组后没人只有一次登录 的对象
                else return null;
            }
        }, new PatternSelectFunction<TimeoutVo, String>() {
            @Override//正常处理函数
            public String select(Map<String, List<TimeoutVo>> map) throws Exception {
                List<TimeoutVo> next = map.get("next");
                if (next != null && next.size() > 0)
                    return next.get(0).name;
                else
                    return null;
            }
        });

        res.print("onTime");
        res.getSideOutput(tag).print("late");

        env.execute();

    }

    public static class TimeoutVo{
        private int id;
        private String name;
        private String op;
        private Long time;

        public TimeoutVo(int id, String name, String op, Long time) {
            this.id = id;
            this.name = name;
            this.op = op;
            this.time = time;
        }

        public int getId() {
            return id;
        }

        public String getName() {
            return name;
        }

        public String getOp() {
            return op;
        }

        public Long getTime() {
            return time;
        }

        public void setId(int id) {
            this.id = id;
        }

        public void setName(String name) {
            this.name = name;
        }

        public void setOp(String op) {
            this.op = op;
        }

        public void setTime(Long time) {
            this.time = time;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (!(o instanceof TimeoutVo)) return false;
            TimeoutVo timeoutVo = (TimeoutVo) o;
            return getId() == timeoutVo.getId() && getName().equals(timeoutVo.getName()) && getOp().equals(timeoutVo.getOp()) && getTime().equals(timeoutVo.getTime());
        }

        @Override
        public int hashCode() {
            return Objects.hash(getId(), getName(), getOp(), getTime());
        }

        @Override
        public String toString() {
            return "TimeoutVo{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    ", op='" + op + '\'' +
                    ", time=" + time +
                    '}';
        }
    }
}
